#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import json

from kafka import KafkaConsumer
from configparser import ConfigParser

from common import mq_config_abs_path, common_abs_path, hlog
from data_handle.lagou_url import LagouUrlClean
from data_handle.zhipin_url import ZhipinUrlClean


def main():
    hlog.enter_func("position_url_html_consumer.main()")

    config = ConfigParser()
    config.read(mq_config_abs_path.absolute(), encoding='UTF-8')
    host = config['mq_positions_url_html']["host"]
    group = config['mq_positions_url_html']["group"]
    topic = config['mq_positions_url_html']["topic"]

    platform_config = ConfigParser()
    platform_config.read(common_abs_path.absolute(), encoding='UTF-8')

    consumer = KafkaConsumer(topic,
                             group_id=group,
                             bootstrap_servers=[host])

    while True:
        if sppm.signal_monitor():
            break

        for message in consumer:

            sppm.working_lock(True)

            json_string = message.value.decode("utf-8")

            try:
                load_dict = json.loads(json_string)

                if load_dict["platform"] == platform_config["platform"]["LaGou"]:
                    result = LagouUrlClean(load_dict)
                    result.data_clean()
                elif load_dict["platform"] == platform_config["platform"]["ZhiPin"]:
                    result = ZhipinUrlClean(load_dict)
                    result.data_clean()
            except:
                pass

            sppm.working_lock(False)


if __name__ == "__main__":
    import sppm

    sppm.sppm_start(main)
